package org.leafDomain.combine.Impl;

import org.apache.hadoop.hbase.client.Put;
import org.apache.log4j.Logger;
import org.leafDomain.combine.DataSourceToHBase;
import org.leafDomain.file.FileSource;
import org.leafDomain.hbase.HBase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class FileSourceToHBaseImpl implements DataSourceToHBase<FileSource> {

    private static final Logger loggerFile = Logger.getLogger(FileSourceToHBaseImpl.class);
    private FileSource fileSource;
    private HBase hbase;

    @Override
    public void setSource(FileSource source) {
        this.fileSource = source;
    }

    @Override
    public void setHBase(HBase hbase) {
        this.hbase = hbase;
    }

    @Override
    public void startTransfer() throws IOException {
        List<Put> container = new ArrayList<>();
        try{
            fileSource.init();
            loggerFile.info("FileSource 初始化成功");
            hbase.init();
            loggerFile.info("HBase 初始化成功");

            loggerFile.info("数据从 FileSource 迁移至 HBase 开始...");

            int count = 0;
            while (fileSource.read(container)) {
                hbase.putBatch(container);
                loggerFile.info(String.format("第 %d 批：%d 条数据插入成功", ++count, container.size()));
            }

            loggerFile.info("数据从 FileSource 迁移至 HBase 开始...");

        }catch (Exception e){
            loggerFile.error("将 FileSource 数据批量迁移至 HBase 异常：",e);
        } finally {
            // 先开的后释放后开的先释放
            hbase.close();
            fileSource.close();
        }

    }
}
